import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author wsl
 * @version 2020-12-29
 */
object WC {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("spark").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))

    ssc.socketTextStream("usdp-o3tbdsfp-monitor1", 4444)
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      // .print()
      .foreachRDD(
        rdd => {

          // 在Driver端执行(ctrl+n JobScheduler)，一个批次一次
          // 在JobScheduler 中查找（ctrl + f）streaming-job-executor
          println("222222:" + Thread.currentThread().getName)

          rdd.foreachPartition(
            //5.1 测试代码
            iter => iter.foreach(println)

            //5.2 企业代码
            //5.2.1 获取连接
            //5.2.2 操作数据，使用连接写库
            //5.2.3 关闭连接
          )

        }
      )


    ssc.start()
    ssc.awaitTermination()

  }
}
